
                /*********************************************
                **                                          **
                **                                          **
                **                ca.stop.C                 **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/

/*
Asynchronous Communication.
Thread hands control over the local scheduler. 
Polling is automatic performed by local scheduler.
*/

#include "arch.h"

Global_id gid1;
Coll_id cid1, cid2;
void init_gid(int, char **){
   gid1.set();
   cid1.set();
   cid2.set();
   fprintf(Sdf::trace, "node%d: gid1.tag=%d, cid1.tag=%d, cid2.tag=%d\n",
                Sdf::PROC_ID, gid1.get(), cid1.get(), cid2.get());
   fflush(Sdf::trace);
}

class SR_count{
   public:
      int cnt;
      SR_count(){
         cnt = 0;
      }
      void reset(){
         cnt = 0;
      }
      void more_SR(int i){
         cnt += i;
      }
};

#define wait_while(condition) while(condition) Sdf::poll();
#define not_complete  (c.cnt >0)

ChanA<int> *chan;

void upon_end_of_SR_r(ChanAs *chan);           
void upon_end_of_SR_s(ChanAs *chan);           

class  Root_thread: public Thread{
   int recvd;
  public:
        Root_thread()
                : Thread((Tmf0_ptr)&Root_thread::body){}

        void body(){
            fprintf(Sdf::trace, "node%d, in Root_thread::body1\n", Sdf::PROC_ID);
            fflush(Sdf::trace);
            chan = new ChanA<int>(gid1, (Sdf::PROC_ID+1)%Sdf::PROC_NB);

            fprintf(Sdf::trace, "node%d, in Root_thread::body2\n", Sdf::PROC_ID);
            fflush(Sdf::trace);

              //this is to ensure global completion of chan construction
              //(barrier is not really neccessary)
            Sdf::barrier(cid1);

            SR_count c;
            chan->set(Sdf::SND, upon_end_of_SR_s, &c);
            chan->set(Sdf::RCV, upon_end_of_SR_r, &c);
            fprintf(Sdf::trace, "node%d, in Root_thread::body3\n", Sdf::PROC_ID);
            fflush(Sdf::trace);
            int sum_recvd = 0;
            for(int i=0; i<5; i++){
               int sent_data = Sdf::PROC_ID*10 + i;
               c.more_SR(2);
               chan->send(Sdf::STD, sent_data);
               chan->recv(recvd);
               stop();
               Sdf::barrier(cid2);
               int reduce_out;
               MPI_Allreduce(&recvd, &reduce_out, 1, MPI_INT,
                       MPI_SUM, MPI_COMM_WORLD);
               sum_recvd += reduce_out;
              fprintf(Sdf::trace, "node%d, i=%d, recvd=%d, reduce_out=%d, sum_recvd=%d\n", 
                      Sdf::PROC_ID, i, recvd, reduce_out, sum_recvd);
              fflush(Sdf::trace);
            }
            delete chan;

       }
};

MAIN(Root_thread, init_gid, Sdf::empty, Sdf::empty)


void upon_end_of_SR_r(ChanAs *chan){

   fprintf(Sdf::trace, "node%d, upon_end_of_SR_r, capa=%d, rcvd size=%d, Tsize=%d\n",
                Sdf::PROC_ID, chan->get_chan_capacity(),
                chan->get_msg_size(Sdf::RCV), chan->get_Tsize());
   fflush(Sdf::trace);
   if(!(((SR_count *)(chan->get_hndlr_data_ptr(Sdf::RCV)))->cnt -= 1))
      Sdf::ROOT_THREAD->schedule();
}

void upon_end_of_SR_s(ChanAs *chan){
   fprintf(Sdf::trace, "node%d, upon_end_of_SR_s, capa=%d, sent size=%d, Tsize=%d\n",
                Sdf::PROC_ID, chan->get_chan_capacity(),
                chan->get_msg_size(Sdf::SND),  chan->get_Tsize());
   fflush(Sdf::trace);
   if(!(((SR_count *)(chan->get_hndlr_data_ptr(Sdf::SND)))->cnt -= 1))
      Sdf::ROOT_THREAD->schedule();
}
